%%%-------------------------------------------------------------------
%%% @author glendy
%%% @copyright (C) 2025, <COMPANY>
%%% @doc
%%%
%%% @end
%%% Created : 16. 9月 2025 8:40
%%%-------------------------------------------------------------------
-module(redis_worker).
-behaviour(gen_server).
-author("glendy").

-include("redis_driver.hrl").

%% API
-export([
  start_link/0,
  start_link/2,
  start_link/3,
  start_link/4,
  start_link/5,
  start_link/6,
  start_link/7,
  init/1,
  handle_call/3,
  handle_cast/2,
  handle_info/2,
  code_change/3,
  terminate/2
]).

-record(state, {
  host :: string() | undefined,
  port :: integer() | undefined,
  password :: binary() | undefined,
  database :: binary() | undefined,
  reconnect_sleep :: reconnect_sleep() | undefined,
  connect_timeout :: integer() | undefined,
  socket_options :: list(),

  socket :: port() | undefined,
  parser_state :: #pstate{} | undefined,
  queue
}).

-define(TIMEOUT, 5000).

start_link() ->
  start_link("127.0.0.1", 6379).

start_link(Host, Port) ->
  start_link(Host, Port, 0).

start_link(Host, Port, Database) ->
  start_link(Host, Port, Database, "").

start_link(Host, Port, Database, Password) ->
  start_link(Host, Port, Database, Password, 100).

start_link(Host, Port, Database, Password, ReconnectSleep) ->
  start_link(Host, Port, Database, Password, ReconnectSleep, ?TIMEOUT).

start_link(Host, Port, Database, Password, ReconnectSleep, ConnectTimeout) ->
  start_link(Host, Port, Database, Password, ReconnectSleep, ConnectTimeout, []).

start_link(Host, Port, Database, Password, ReconnectSleep, ConnectTimeout, SocketOptions)
  when is_list(Host) orelse
            (is_tuple(Host) andalso tuple_size(Host) =:= 2 andalso element(1, Host) =:= local),
      is_integer(Port),
      is_integer(Database) orelse Database == undefined,
      is_list(Password),
      is_integer(ReconnectSleep) orelse ReconnectSleep =:= no_reconnect,
      is_integer(ConnectTimeout),
      is_list(SocketOptions) ->
  gen_server:start_link(?MODULE, [Host, Port, Database, Password,
    ReconnectSleep, ConnectTimeout, SocketOptions], []).

init([Host, Port, Database, Password, ReconnectSleep, ConnectTimeout, SocketOptions]) ->
  State = #state{host = Host,
    port = Port,
    database = read_database(Database),
    password = list_to_binary(Password),
    reconnect_sleep = ReconnectSleep,
    connect_timeout = ConnectTimeout,
    socket_options = SocketOptions,

    parser_state = redis_decode:init(),
    queue = queue:new()
  },

  case ReconnectSleep of
    no_reconnect ->
      case connect(State) of
        {ok, _NewState} = Res -> Res;
        {error, Reason} -> {stop, Reason}
      end;
    T when is_integer(T) ->
      self() ! try_req_connect,
      {ok, State}
  end.

read_database(undefined) ->
  undefined;
read_database(Database) when is_integer(Database) ->
  list_to_binary(integer_to_list(Database)).

connect(State) ->
  {ok, {AFamily, Addr}} = redis_util:get_addr(State#state.host),
  Port = case AFamily of
           local -> 0;
           _ -> State#state.port
         end,

  SocketOptions = lists:ukeymerge(1, lists:keysort(1, State#state.socket_options), lists:keysort(1, ?SOCKET_OPTS)),
  ConnectOptions = [AFamily | [?SOCKET_MODE | SocketOptions]],

  case gen_tcp:connect(Addr, Port, ConnectOptions, State#state.connect_timeout) of
    {ok, Socket} ->
      case authenticate(Socket, State#state.password) of
        ok ->
          case select_database(Socket, State#state.database) of
            ok ->
              ok = inet:setopts(Socket, [{active, true}]),
              {ok, State#state{socket = Socket}};
            {error, Reason} ->
              {error, {select_error, Reason}}
          end;
        {error, Reason} ->
          {error, {authentication_error, Reason}}
      end;
    {error, Reason} ->
      {error, {connection_error, Reason}}
  end.

maybe_reconnect(Reason, #state{reconnect_sleep = no_reconnect, queue = Queue} = State) ->
  reply_all({error, Reason}, Queue),
  {stop, normal, State#state{socket = undefined}};
maybe_reconnect(Reason, #state{reconnect_sleep = ReconnectSleep, queue = Queue} = State) ->
  error_logger:error_msg("eredis: Re-establishing connection to ~p:~p due to ~p",
    [State#state.host, State#state.port, Reason]),
  erlang:send_after(ReconnectSleep, self(), try_req_connect),
  reply_all({error, Reason}, Queue),
  {noreply, State#state{socket = undefined, queue = queue:new()}}.

select_database(_Socket, undefined) ->
  ok;
select_database(_Socket, <<"0">>) ->
  ok;
select_database(Socket, Database) ->
  DataBaseBin = redis_util:to_binary(Database),
  do_sync_command(Socket, ?REDIS_CMD_SELECT(DataBaseBin)).

authenticate(_Socket, <<>>) ->
  ok;
authenticate(Socket, Password) ->
  PasswordBin = redis_util:to_binary(Password),
  do_sync_command(Socket, ?REDIS_CMD_AUTH(PasswordBin)).

%% return "+OK\r\n", otherwise it will fail.
do_sync_command(Socket, Command) ->
  ok = inet:setopts(Socket, [{active, false}]),
  case gen_tcp:send(Socket, Command) of
    ok ->
      %% Hope there's nothing else coming down on the socket..
      case gen_tcp:recv(Socket, 0, ?RECV_TIMEOUT) of
        {ok, <<"+OK\r\n">>} ->
          ok;
        Other ->
          {error, {unexpected_data, Other}}
      end;
    {error, Reason} ->
      {error, Reason}
  end.

reply(Value, Queue) ->
  case queue:out(Queue) of
    {{value, {1, From}}, NewQueue} ->
      safe_reply(From, Value),
      NewQueue;
    {{value, {1, From, Replies}}, NewQueue} ->
      safe_reply(From, lists:reverse([Value | Replies])),
      NewQueue;
    {{value, {N, From, Replies}}, NewQueue} when N > 1 ->
      queue:in({N - 1, From, [Value | Replies]}, NewQueue);
    {empty, Queue} ->
      %% Oops
      error_logger:info_msg("eredis: Nothing in queue, but got value from parser~n"),
      exit(empty_queue)
  end.

reply_all(Value, Queue) ->
  case queue:peek(Queue) of
    empty ->
      Queue;
    {value, {_, From}} ->
      safe_reply(From, Value),
      reply_all(Value, queue:drop(Queue));
    {value, {_, From, _Req}} ->
      safe_reply(From, Value),
      reply_all(Value, queue:drop(Queue))
  end.

safe_reply(undefined, _Value) ->
  ok;
safe_reply(Pid, Value) when is_pid(Pid) ->
  safe_send(Pid, {response, Value});
safe_reply(From, Value) ->
  gen_server:reply(From, Value).

safe_send(Pid, Value) ->
  try
    erlang:send(Pid, Value)
  catch
    Err:Reason ->
      error_logger:info_msg("eredis: Failed to send message to ~p with reason ~p~n", [Pid, {Err, Reason}])
  end.

do_request(_Req, _From, #state{socket = undefined} = State) ->
  {reply, {error, no_connection}, State};
do_request(Req, From, State) ->
  case gen_tcp:send(State#state.socket, Req) of
    ok ->
      NewQueue = queue:in_r({1, From}, State#state.queue),
      {noreply, State#state{queue = NewQueue}};
    {error, Reason} ->
      {reply, {error, Reason}, State}
  end.

do_pipeline(_Pipeline, _From, #state{socket = undefined} = State) ->
  {reply, {error, no_connection}, State};
do_pipeline(Pipeline, From, State) ->
  case gen_tcp:send(State#state.socket, Pipeline) of
    ok ->
      NewQueue = queue:in_r({length(Pipeline), From, []}, State#state.queue),
      {noreply, State#state{queue = NewQueue}};
    {error, Reason} ->
      {reply, {error, Reason}, State}
  end.

do_tcp_response(Data, #state{socket = Socket, parser_state = ParserState, queue = Queue} = State) ->
  RestBin = ParserState#pstate.rest_bin,
  NewParserState0 = ParserState#pstate{
    rest_bin = <<RestBin/binary, Data/binary>>
  },
  case redis_decode:decode(NewParserState0) of
    {ok, Value, NewParserState} ->
      NewQueue = reply({ok, Value}, Queue),
      State#state{parser_state = NewParserState, queue = NewQueue};
    {continue, NewParserState} ->
      State#state{parser_state = NewParserState};
    {error, Reason} ->
      error_logger:error_msg("redis: Error parsing response from Redis: ~p~n", [Reason]),
      NewQueue = reply_all({error, Reason}, Queue),
      if
        Socket =/= undefined ->
          gen_tcp:close(Socket);
        true ->
          ok
      end,
      self() ! try_req_connect,
      State#state{socket = undefined, parser_state = redis_decode:init(), queue = NewQueue}
  end.

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
handle_call({request, Req}, From, State) ->
  do_request(Req, From, State);
handle_call({pipeline, Pipeline}, From, State) ->
  do_pipeline(Pipeline, From, State);
handle_call(stop, _From, State) ->
  {stop, normal, ok, State};
handle_call(_Request, _From, State) ->
  {reply, unknown_request, State}.

handle_cast({request, Req}, State) ->
  case do_request(Req, undefined, State) of
    {reply, _Reply, State1} ->
      {noreply, State1};
    {noreply, State1} ->
      {noreply, State1}
  end;
handle_cast({request, Req, Pid}, State) ->
  case do_request(Req, Pid, State) of
    {reply, Reply, State1} ->
      safe_send(Pid, {response, Reply}),
      {noreply, State1};
    {noreply, State1} ->
      {noreply, State1}
  end;
handle_cast(_Msg, State) ->
  {noreply, State}.

handle_info({tcp, Socket, Bs}, #state{socket = Socket} = State) ->
  ok = inet:setopts(Socket, [{active, once}]),
  {noreply, do_tcp_response(Bs, State)};
handle_info({tcp, Socket, _}, #state{socket = OurSocket} = State)
  when OurSocket =/= Socket ->
  {noreply, State};
handle_info({tcp_error, _Socket, _Reason}, State) ->
  {noreply, State};
handle_info({tcp_closed, _Socket}, State) ->
  maybe_reconnect(tcp_closed, State);
handle_info({connection_ready, Socket}, #state{socket = undefined} = State) ->
  {noreply, State#state{socket = Socket}};
handle_info(stop, State) ->
  {stop, shutdown, State};
handle_info(try_req_connect, #state{socket = undefined} = State) ->
  case connect(State) of
    {ok, NewState} ->
      {noreply, NewState};
    {error, Reason} ->
      maybe_reconnect(Reason, State)
  end;
handle_info(_Info, State) ->
  {stop, {unhandled_message, _Info}, State}.

code_change(_OldVsn, State, _Extra) ->
  {ok, State}.

terminate(_Reason, State) ->
  case State#state.socket of
    undefined -> ok;
    Socket -> gen_tcp:close(Socket)
  end,
  ok.
